package com.cn.file;

import com.cn.exception.MyException;
import com.cn.file.ha.HAService;

import java.nio.ByteBuffer;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 消息实际存储
 */
public class CommitLog {

    private MapedFileQueue mapedFileQueue;
    private final DefaultMessageStore defaultMessageStore;
    private final ReentrantLock writeLock;
    private FlushCommitLogService flushCommitLogService;
    private AppendMessageCallback appendMessageCallback;
    private boolean isOk = false;
    private Map<String, Long> topicQueueTable = new HashMap<>(1024);
    private final HAService haService;

    public CommitLog(DefaultMessageStore defaultMessageStore) {
        this.defaultMessageStore = defaultMessageStore;
        this.mapedFileQueue = new MapedFileQueue(this.defaultMessageStore.getMessageStoreConfig().getStorePath(),
                this.defaultMessageStore.getMessageStoreConfig().getMappedFileSize());
        if (this.defaultMessageStore.getMessageStoreConfig().getFlushDiskType().equals("sync")) {
            this.flushCommitLogService = new GroupCommitService();
        } else {
            this.flushCommitLogService = new FlushRealCommitService();
        }
        this.writeLock = new ReentrantLock();
        this.appendMessageCallback = new DefaultAppendMessageCallback(this.defaultMessageStore.getMessageStoreConfig().getMaxMessageSize());
        this.haService = new HAService(this.defaultMessageStore, this);
    }

    public void load() {
        this.mapedFileQueue.load();
        this.isOk = true;
    }

    /**
     * 恢复commitlog文件的可写入位置
     */
    public void recover() {
        //以下代码是暂时性的，不够严谨
        this.mapedFileQueue.recover(0);
    }

    public void start() {
        this.flushCommitLogService.start();
    }

    public void stop() {
        this.flushCommitLogService.shutDown();
    }


    public Map<String, Long> getTopicQueueTable() {
        return topicQueueTable;
    }

    public void setTopicQueueTable(Map<String, Long> topicQueueTable) {
        this.topicQueueTable = topicQueueTable;
    }

    /**
     * 当前数据文件的可写入位置
     *
     * @return
     */
    public long getMaxOffset() {
        return this.mapedFileQueue.getMaxOffset();
    }

    public SelectMappedBufferResult getData(long offset) {
        return this.getData(offset, offset == 0);
    }

    public SelectMappedBufferResult getData(long offset, int msgSize) {
        //commitlog数据文件大小
        int size = this.defaultMessageStore.getMessageStoreConfig().getMappedFileSize();
        MappFile mappFile = this.mapedFileQueue.getLastMappedFile(offset, false);
        if (mappFile != null) {
            //计算读取数据的下标位置
            int pos = (int) offset % size;
            SelectMappedBufferResult selectMappedBufferResult = mappFile.selectMappedBuffer(pos, msgSize);
            return selectMappedBufferResult;
        }
        return null;
    }

    public DispatchRequest checkMessageAndReturnSize(ByteBuffer byteBuffer, boolean checkCRC,
                                                     boolean readBody) {
        try {
            // 1. 消息总长度
            int totalSize = byteBuffer.getInt();//4位TOTALSIZE

            // 2. topic长度
            byte topicLength = byteBuffer.get();

            //3. topic内容
            byte[] topics = new byte[topicLength];
            byteBuffer.get(topics);
            String topic = new String(topics, Charset.forName("utf-8"));

            //4. queueId
            int queueId = byteBuffer.getInt();

            //5. 第几个消息（相对于queue队列）
            long queueOffset = byteBuffer.getLong();


            //6. 物理位置
            long physicaloffset = byteBuffer.getLong();

            //7. 消息
            int bodyLength = byteBuffer.getInt();//4位
            if (bodyLength > 0) {
                byte[] datas = new byte[bodyLength];//totalsize
                byteBuffer.get(datas);//消息内容

            }

            //7. 存储时间
            long storeTimestamp = byteBuffer.getLong();

            return new DispatchRequest(topic, queueId, physicaloffset, totalSize, storeTimestamp, queueOffset);
        } catch (Exception e) {

        }
        return null;
    }

    /**
     * 通过pos来获取数据文件块
     *
     * @param offset
     * @param returnFirstOnNotFound
     * @return
     */
    public SelectMappedBufferResult getData(long offset, boolean returnFirstOnNotFound) {
        //commitlog数据文件可存放数据大小
        int size = this.defaultMessageStore.getMessageStoreConfig().getMappedFileSize();
        MappFile mappFile = this.mapedFileQueue.getLastMappedFile(offset, returnFirstOnNotFound);
        if (mappFile != null) {
            /**
             * 计算实际位移
             * 1. 假设文件大小1024；
             * 2. offset =10，那么实际位移=10
             * 3. offset = 1096，那么实际位移=72
             */
            int pos = (int) offset % size;
            SelectMappedBufferResult selectMappedBufferResult = mappFile.selectMappedBuffer(pos);
            return selectMappedBufferResult;
        }
        return null;
    }

    public AppendMessageResult putMessage(MessageStore messageStore) {
        if (!this.isOk) throw new MyException("服务未启动！");
        if (messageStore == null) throw new MyException("消息不能为空");
        if (messageStore.getBody() == null || messageStore.getBody().length == 0) throw new MyException("消息不能为空");
        try {
            this.writeLock.lock();
            try {
                MappFile mappFile = this.mapedFileQueue.getLastMappedFile();
                if (mappFile == null) {
                    mappFile = this.mapedFileQueue.getLastMappedFile(0, true);
                }
                if (mappFile == null) {
                    return new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR, 0, 0, null, 0, 0, 0, 0);
                }

                AppendMessageResult result = mappFile.doAppend(messageStore, appendMessageCallback);

                //刷盘
                handleDiskFlush();

                //同步到从节点
                handleHaFlush(messageStore, result);

                return result;
            } finally {
                this.writeLock.unlock();
            }
        } catch (Exception e) {
//            e.printStackTrace();
        }

        return new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR, 0, 0, null, 0, 0, 0, 0);
    }

    public void handleHaFlush(MessageStore messageStore, AppendMessageResult result) {
        if (true) {
            return;
        }
        if (result.getStatus() == AppendMessageStatus.PUT_OK) {
            long offset = result.getWroteOffset() + result.getWroteBytes();
            //判断slave复制端的位移是否小于当前位移
            if (true) {
                GroupCommitRequest request = new GroupCommitRequest(offset);
                haService.addRequest(request);

                //等待500毫秒
                boolean flush = request.waitForFlush(500);
                if (flush) {
                    //复制成功
                    System.out.println("复制成功");

                } else {
                    //复制失败
                    System.out.println("复制超时");
                }
            }
        }
    }

    public void handleDiskFlush() {
        if (this.defaultMessageStore.getMessageStoreConfig().getFlushDiskType().equals("sync")) {
            //同步刷新，写入一条，刷盘一条
            GroupCommitService groupCommitService = (GroupCommitService) this.flushCommitLogService;
            //nextOffset暂时为0
            GroupCommitRequest request = new GroupCommitRequest(0);
            groupCommitService.putRequest(request);
            request.waitForFlush(1000);//超时时间
        } else {
            //异步刷盘
            this.flushCommitLogService.wakeup();
        }
    }

    /**
     * 计数器（防止长时间等待）
     */
    public static class GroupCommitRequest {
        private final long nextOffset;
        private final CountDownLatch countDownLatch = new CountDownLatch(1);
        //false-刷盘失败 true-刷盘成功
        private volatile boolean flushOK = false;

        public GroupCommitRequest(long nextOffset) {
            this.nextOffset = nextOffset;
        }

        public long getNextOffset() {
            return nextOffset;
        }

        public void wakeupCustomer(boolean flushOK) {
            this.flushOK = flushOK;
            this.countDownLatch.countDown();
        }

        public boolean waitForFlush(long timeout) {
            try {
                //等待超时时间
                this.countDownLatch.await(timeout, TimeUnit.MILLISECONDS);
                return this.flushOK;
            } catch (InterruptedException e) {
                e.printStackTrace();
                return false;
            }
        }

    }

    class GroupCommitService extends FlushCommitLogService {
        //写队列
        private volatile List<GroupCommitRequest> requestWrite = new ArrayList<>();
        //读队列
        private volatile List<GroupCommitRequest> requestRead = new ArrayList<>();

        //写入数据
        public synchronized void putRequest(GroupCommitRequest request) {
            synchronized (this.requestWrite) {
                this.requestWrite.add(request);
            }
            //有新的消息来了，可以刷盘了
            if (hasNotified.compareAndSet(false, true)) {
                waitPoint.countDown();//notify
            }
        }

        //交换缓存
        public void swapRequests() {
            List<GroupCommitRequest> tmp = this.requestRead;
            this.requestRead = this.requestWrite;
            this.requestWrite = tmp;
        }


        @Override
        public String getServiceName() {
            return this.getClass().getSimpleName();
        }

        /**
         * 刷盘
         */
        private void doCommit() {
            synchronized (this.requestRead) {
                if (!this.requestRead.isEmpty()) {
                    //遍历等待刷盘的数据
                    for (GroupCommitRequest request : this.requestRead) {
                        boolean flushOk = false;
                        flushOk = mapedFileQueue.flush(0);
                        //通知进程，刷盘完成
                        request.wakeupCustomer(flushOk);
                    }
                    this.requestRead.clear();
                } else {
                    this.requestRead.clear();
                }
            }
        }

        @Override
        public void run() {
            while (!this.isStopped()) {
                this.waitForRunning(30);
                this.doCommit();
            }
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            synchronized (this) {
                swapRequests();
            }
            this.doCommit();

        }

        @Override
        protected void onWaitEnd() {
            swapRequests();
        }
    }

    //异步刷盘
    class FlushRealCommitService extends FlushCommitLogService {

        private long lastFlushTimestamp = 0;


        @Override
        public String getServiceName() {
            return this.getClass().getName();
        }

        @Override
        public void run() {
            while (!isStopped()) {
                int interval = defaultMessageStore.getMessageStoreConfig().getFlushIntervalCommitLog();
                int flushLeastPages = defaultMessageStore.getMessageStoreConfig().getFlushLeastPages();

                //当前时间
                long currentTimeMillis = System.currentTimeMillis();
                if (currentTimeMillis >= (lastFlushTimestamp + defaultMessageStore.getMessageStoreConfig().getFlushCommitLogThoroughInterval())) {
                    this.lastFlushTimestamp = currentTimeMillis;
                    flushLeastPages = 0;
                }
                try {
                    this.waitForRunning(interval);
                    mapedFileQueue.flush(flushLeastPages);
                } catch (Exception e) {

                }
            }
        }
    }

    class DefaultAppendMessageCallback implements AppendMessageCallback {

        private final ByteBuffer msgIdMemory;

        // Store the message content
        private final ByteBuffer msgStoreItemMemory;

        private final ByteBuffer hostHolder = ByteBuffer.allocate(8);

        public DefaultAppendMessageCallback(int size) {
            //处理消息id
            this.msgIdMemory = ByteBuffer.allocate(16);
            //最多存储4k大小的消息
            this.msgStoreItemMemory = ByteBuffer.allocate(1024 * 4);
        }

        @Override
        public AppendMessageResult doAppend(long fileFromOffset, ByteBuffer byteBuffer, int maxBlank, MessageStore msg) {
            if (byteBuffer == null) {
                return new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR, 0, 0, null, 0, 0, 0, 0);
            }
            try {
                //该消息写入的位置
                long wrotePosition = fileFromOffset + byteBuffer.position();

                this.resetByteBuffer(hostHolder, 8);

                //存储时间
                msg.setStoreTimestamp(System.currentTimeMillis());

                //该消息要存储的时间长度
                //主题序列化
                final byte[] topicData = msg.getTopic().getBytes(MessageUtil.CHARSET_UTF8);
                //主题大小
                final int topicLength = topicData.length;
                //消息大小
                final int bodyLength = msg.getBody() == null ? 0 : msg.getBody().length;

                //消息实际大小
                final int msgLen = calMsgLength(bodyLength, topicLength);

                this.resetByteBuffer(msgStoreItemMemory, msgLen);
                //1.消息实际大小
                this.msgStoreItemMemory.putInt(msgLen);
                //2.topic
                this.msgStoreItemMemory.put((byte) topicLength);
                this.msgStoreItemMemory.put(topicData);
                //3.queueId
                this.msgStoreItemMemory.putInt(msg.getQueueId());

                String topKey = msg.getTopic() + "-" + msg.getQueueId();
                Long queueOffset = CommitLog.this.topicQueueTable.get(topKey);
                if (queueOffset == null) {
                    queueOffset = 0l;
                }
                //4.该消息表示在队列中的第几条
                this.msgStoreItemMemory.putLong(queueOffset.longValue());

                //5.消息存储在文件中的物理位置
                this.msgStoreItemMemory.putLong(wrotePosition);

                //6.msg
                this.msgStoreItemMemory.putInt(bodyLength);
                if (bodyLength > 0) {
                    this.msgStoreItemMemory.put(msg.getBody());
                }
                //6.存储时间
                this.msgStoreItemMemory.putLong(msg.getStoreTimestamp());

                byteBuffer.put(this.msgStoreItemMemory.array(), 0, msgLen);

                CommitLog.this.topicQueueTable.put(topKey, ++queueOffset);

                return new AppendMessageResult(AppendMessageStatus.PUT_OK, wrotePosition, msgLen, msg.getMsgId(), msg.getStoreTimestamp(), 0, 0, 0);

            } catch (Exception e) {
                return new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR, 0, 0, msg.getMsgId(), msg.getStoreTimestamp(), 0, 0, 0);
            }
        }

        /**
         * 重置buffer
         *
         * @param byteBuffer
         * @param limit
         */
        private void resetByteBuffer(final ByteBuffer byteBuffer, final int limit) {
            byteBuffer.flip();//重置下标，从0的位置开始
            byteBuffer.limit(limit);//设置buffer可写入大小
        }

        private int calMsgLength(int bodyLength, int topicLength) {
            final int msgLen = 4 //存放OTALSIZE
                    + 1//存放topiclength
                    + topicLength//存放topic长度
                    + 4 //存放queueid
                    + 8 //存放queueoffset
                    + 8 //物理位置
                    + 4//存放 bodylengh
                    + bodyLength
                    + 8//存储时间
                    ;
            return msgLen;
        }

    }


}


